use mysql::prelude::*;
use mysql::*;
use std::{collections::HashMap, sync::Arc, thread};
use tokio::{runtime, sync::Mutex};

pub const TYPE_SUPPORT: [&str; 34] = [
    "bit",
    "tinyint",
    "bool",
    "boolean",
    "smallint",
    "mediumint",
    "int",
    "integer",
    "bigint",
    "decimal",
    "dec",
    "float",
    "double",
    "double precision",
    "date",
    "datetime",
    "timestamp",
    "time",
    "year",
    "char",
    "varchar",
    "binary",
    "varbinary",
    "tinyblob",
    "tinytext",
    "blob",
    "text",
    "mediumblob",
    "mediumtext",
    "longblob",
    "longtext",
    "enum",
    "set",
    "json",
];

#[derive(Debug, Default, Clone)]
pub struct Table {
    // 所属用户
    pub table_schema: String,
    // 表名
    pub table_name: String,
    // 唯一索引，除去pk
    pub uk: Option<String>,
    // 唯一索引，可空唯一索引
    pub nullable_uk: Option<String>,
    // 主键
    pub pk: Option<String>,
    // 外键
    pub fk: Option<String>,
    // 自身触发器
    pub triggers: Option<String>,
    // 关联的触发器
    pub partitioned: bool,
    // 压缩表：包括行压缩以及页压缩
    pub compressed: bool,
    // 存储引擎
    pub engine: String,
    // 不支持的类型
    pub data_type_unsupport: Option<String>,
    // 有限支持
    pub data_type_support_limited: Option<String>,
    // 虚拟列
    pub virtual_columns: Option<String>,
    // 是否为加密表
    pub encryption: bool,
}

fn schema_table_count(conn: &mut mysql::PooledConn, schema: &str) -> usize {
    let result: Option<usize> = conn
        .query_first(format!(
            "select count(*) from information_schema.tables where table_schema = '{schema}'"
        ))
        .unwrap();
    result.unwrap()
}

fn query_table_attr(
    conn: &mut mysql::PooledConn,
    schema: &String,
    keys: &Vec<String>,
    db_schema_tables: &mut Vec<Table>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    show_nullable_uk: bool,
) {
    // 主键
    let sql = format!("select TABLE_SCHEMA,TABLE_NAME from information_schema.TABLE_CONSTRAINTS where TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}') and CONSTRAINT_TYPE = 'PRIMARY KEY'", keys.join("','"));
    conn.query_map(&sql, |(table_schema, table_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                table.pk = Some(String::from("PRIMARY"));
                break;
            }
        }
    })
    .unwrap();

    // 分区表
    let sql = format!("select TABLE_SCHEMA,TABLE_NAME from information_schema.PARTITIONS where TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}')", keys.join("','"));
    conn.query_map(&sql, |(table_schema, table_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                table.partitioned = true;
                break;
            }
        }
    })
    .unwrap();

    // 唯一索引：排除主键
    let sql = format!("select table_schema,table_name,index_name from information_schema.STATISTICS where TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}') and INDEX_NAME != 'PRIMARY' and NON_UNIQUE = 0 group by table_schema,table_name,index_name", keys.join("','"));
    // println!("sql:{sql}");
    conn.query_map(&sql, |(table_schema, table_name, index_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        let index_name = from_value::<String>(index_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                if let Some(uk) = table.uk.clone() {
                    table.uk = Some(format!("{},{}", uk, index_name));
                } else {
                    table.uk = Some(index_name);
                }
                break;
            }
        }
    })
    .unwrap();

    // 可空唯一索引
    let sql = format!("select table_schema,table_name,index_name from information_schema.STATISTICS where TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}') and INDEX_NAME != 'PRIMARY' and nullable = 'YES' and NON_UNIQUE = 0 group by table_schema,table_name,index_name", keys.join("','"));
    // println!("sql:{sql}");
    conn.query_map(&sql, |(table_schema, table_name, index_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        let index_name = from_value::<String>(index_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                table.nullable_uk = if let Some(uk) = table.nullable_uk.clone() {
                    Some(format!("{},{},", uk, index_name))
                } else {
                    Some(format!("{},", index_name))
                };

                // show_nullable_uk == false
                if !show_nullable_uk {
                    if let Some(uk) = table.uk.clone() {
                        table.uk = Some(uk.replace(index_name.as_str(), "").replace(",,", ",").trim_start_matches(",").trim_end_matches(",").to_string());
                    }
                }
                break;
            }
        }
    })
    .unwrap();

    // 外键
    let sql = format!("SELECT TABLE_SCHEMA, TABLE_NAME, group_concat(CONSTRAINT_NAME) CONSTRAINT_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}') and REFERENCED_TABLE_SCHEMA is not null and REFERENCED_TABLE_NAME is not null group by TABLE_SCHEMA, TABLE_NAME", keys.join("','"));
    conn.query_map(&sql, |(table_schema, table_name, constraint_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        let constraint_name = from_value::<String>(constraint_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                table.fk = Some(constraint_name);
                break;
            }
        }
    })
    .unwrap();

    // 触发器
    let sql = format!("select EVENT_OBJECT_SCHEMA, EVENT_OBJECT_TABLE, group_concat(TRIGGER_NAME) TRIGGER_NAME from information_schema.TRIGGERS where EVENT_OBJECT_SCHEMA = '{schema}' and EVENT_OBJECT_TABLE in ('{}') group by EVENT_OBJECT_SCHEMA, EVENT_OBJECT_TABLE", keys.join("','"));
    conn.query_map(&sql, |(table_schema, table_name, trigger_name)| {
        let table_schema = from_value::<String>(table_schema);
        let table_name = from_value::<String>(table_name);
        let trigger_name = from_value::<String>(trigger_name);
        for table in db_schema_tables.iter_mut() {
            if table.table_schema == table_schema && table.table_name == table_name {
                table.triggers = Some(trigger_name);
                break;
            }
        }
    })
    .unwrap();

    // 列支持情况
    let sql = format!("select table_schema,table_name,column_name,data_type,extra from information_schema.columns where TABLE_SCHEMA = '{schema}' and TABLE_NAME in ('{}') and (DATA_TYPE not in ('{}') or EXTRA = 'VIRTUAL GENERATED')", keys.join("','"), data_type_support.join("','").to_lowercase());
    // println!("sql:{sql}");
    conn.query_map(
        &sql,
        |(table_schema, table_name, column_name, data_type, extra)| {
            let table_schema = from_value::<String>(table_schema);
            let table_name = from_value::<String>(table_name);
            let column_name = from_value::<String>(column_name);
            let data_type = from_value::<String>(data_type);
            let extra = from_value::<Option<String>>(extra);
            for table in db_schema_tables.iter_mut() {
                if table.table_schema == table_schema && table.table_name == table_name {
                    if extra == Some(String::from("VIRTUAL GENERATED")) {
                        // 虚拟列
                        // 用逗号分割
                        if let Some(vc) = table.virtual_columns.clone() {
                            table.virtual_columns = Some(format!("{}{},", vc, column_name));
                        } else {
                            table.virtual_columns = Some(format!("{},", column_name));
                        }
                    }

                    // 再次匹配过滤
                    // 存在虚拟列时会把列类型带回
                    if data_type_support.contains(&data_type) {
                        continue;
                    }

                    // 有限支持
                    if let Some(data_type_support_limited) = data_type_support_limited {
                        if data_type_support_limited.contains(&data_type) {
                            if let Some(typ) = table.data_type_support_limited.clone() {
                                table.data_type_support_limited =
                                    Some(format!("{}{}:{},", typ, column_name, data_type.clone()));
                            } else {
                                table.data_type_support_limited =
                                    Some(format!("{}:{},", column_name, data_type.clone()));
                            }
                            continue;
                        }
                    }

                    // 不支持
                    if let Some(typ) = table.data_type_unsupport.clone() {
                        table.data_type_unsupport =
                            Some(format!("{}{}:{},", typ, column_name, data_type.clone()));
                    } else {
                        table.data_type_unsupport =
                            Some(format!("{}:{},", column_name, data_type.clone()));
                    }

                    break;
                }
            }
        },
    )
    .unwrap();
}

fn inner_query_tables(
    pool: &mysql::Pool,
    schema: &String,
    keys: &Vec<String>,
    build_in_tables: &Vec<String>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    page: usize,
    page_size: usize,
    show_nullable_uk: bool,
) -> Vec<Table> {
    let sql = if keys.len() > 0 {
        format!(
            "select table_schema, table_name, engine, create_options, row_format from information_schema.tables where table_schema = '{schema}' and table_name in ('{}') and TABLE_TYPE = 'BASE TABLE'",
            keys.join("','")
        )
    } else {
        // let low = (page - 1) * page_size + 1;
        // let high = page * page_size;
        // 数据量不大，直接使用offset limit语法糖
        let offset = (page - 1) * page_size;
        let limit = page_size;
        format!(
            "select table_schema, table_name, engine, create_options, row_format from information_schema.tables where table_schema = '{schema}' and TABLE_TYPE = 'BASE TABLE' limit {offset},{limit}"
        )
    };

    println!("[{:?}] =>sql_text: {sql}", thread::current().id());
    let mut keys = Vec::new();
    let mut conn = pool.get_conn().unwrap();
    let mut result = conn.query_iter(&sql).unwrap();
    // println!("sql:{}", sql);
    let mut db_schema_tables = vec![];
    while let Some(result_set) = result.iter() {
        for row in result_set {
            let row = from_row::<(String, String, String, Option<String>, Option<String>)>(row.unwrap());
            let table_schema = row.0;
            let table_name = row.1;

            let schema_table = format!("{}.{}", table_schema, table_name);
            if build_in_tables.contains(&schema_table) {
                println!("--- Build-in Table: {}", schema_table);
                continue;
            }

            let mut table = Table::default();
            table.table_schema = table_schema;
            table.table_name = table_name;
            table.engine = row.2;
            if let Some(table_options) = row.3 {
                let table_options = table_options.to_uppercase();
                if table_options.contains("ENCRYPTION=\"Y\""){
                    table.encryption = true;
                }
                // Enabling Page Compression
                // COMPRESSION [=] {'ZLIB' | 'LZ4' | 'NONE'}
                if table_options.contains("COMPRESSION=\"ZLIB\"") || table_options.contains("COMPRESSION=\"LZ4\""){
                    table.compressed = true;
                }
            }
            let row_format = row.4;
            if let Some(row_format) = row_format {
                match row_format.as_str() {
                    "Compressed" => {
                        table.compressed = true;
                    }
                    _ => {}
                }
            }

            keys.push(table.table_name.clone());
            db_schema_tables.push(table);
        }
    }

    // 查询表其他属性
    let mut conn = pool.get_conn().unwrap();
    query_table_attr(
        &mut conn,
        &schema,
        &keys,
        &mut db_schema_tables,
        data_type_support,
        data_type_support_limited,
        show_nullable_uk,
    );

    // println!("db_schema_tables: {:?}", db_schema_tables);

    db_schema_tables
}

pub async fn query_tables(
    pool: &mysql::Pool,
    schema_tables: &HashMap<String, Vec<String>>,
    schemas: &Vec<String>,
    build_in_tables: &Vec<String>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    page_size: usize,
    show_nullable_uk: bool,
) -> (HashMap<String, Vec<Table>>, usize) {
    let mut table_count = 0;

    let mut db_user_all_tables = HashMap::new();

    for schema in schemas.clone() {
        let mut tables = match schema_tables.get(&schema) {
            Some(tables) => tables.to_vec(),
            None => continue,
        };

        let mut db_user_tables: Vec<Table> = Vec::new();
        let rt = runtime::Builder::new_multi_thread()
            .worker_threads(4)
            .enable_io()
            .enable_time()
            .build()
            .unwrap();

        let mut count = tables.len();
        if count == 1 && tables[0].is_empty() {
            // User ""
            // 查询整个用户的情况，需要统计用户的所有表
            let mut conn = pool.get_conn().unwrap();
            count = schema_table_count(&mut conn, &schema);
            tables.clear();
        }

        // 启用多线程
        let mut handles = vec![];
        let page_count = count / page_size + if count % page_size == 0 { 0 } else { 1 };
        // println!("page_count: {}", page_count);
        let page = Arc::new(Mutex::new(page_count));
        loop {
            let mut lock = page.lock().await;
            let current_page: usize = lock.to_string().parse().unwrap();
            *lock -= 1;

            let mut cloned_tables = tables.clone();
            tables = cloned_tables.split_off(if cloned_tables.len() > page_size {
                page_size
            } else {
                cloned_tables.len()
            });

            let cloned_build_in_tables = build_in_tables.clone();
            let cloned_schema = schema.clone();
            let cloned_data_type_support = data_type_support.clone();
            let cloned_data_type_support_limited = data_type_support_limited.clone();
            let cloned_pool = pool.clone();
            let handle = rt.spawn(async move {
                inner_query_tables(
                    &cloned_pool,
                    &cloned_schema,
                    &cloned_tables,
                    &cloned_build_in_tables,
                    &cloned_data_type_support,
                    &cloned_data_type_support_limited,
                    current_page,
                    page_size,
                    show_nullable_uk,
                )
                // query(&conn, &cloned_user_list).await
            });
            handles.push(handle);

            if current_page <= 1 {
                break;
            }
        }

        for handle in handles {
            let mut tables = handle.await.unwrap();
            table_count += tables.len();
            db_user_tables.append(&mut tables);
        }
        rt.shutdown_background();

        db_user_all_tables.insert(schema.clone(), db_user_tables);
    }

    (db_user_all_tables, table_count)
}
